Part 2 of #214, on top of the consolidated `StorageBase` from #215–#217.

Users can now declare their own schema instead of picking one of the
three built-in ones — while `add` and the WAL merge behave identically
to `RolloutStore`, because they are literally the same code path, not a
reimplementation.

## Three pieces

### `schema_spec` — schema as a value

The built-in stores hard-code an Arrow schema and then repeat its field
list across five to seven places (schema fn, builders, decoder,
projection, filters, merge key, index name). A `SchemaSpec` is declared
once at creation and persisted with the store.

Validation enforces what the storage layer actually assumes:

- **`id` is mandatory** — `Utf8`, non-nullable, tagged
`unenforced-primary-key`. It is always the LSM merge key and the indexed
column, which is what makes a retried append idempotent and a crashed
merge safe to redo.
- **Reserved names rejected** — anything starting with `_`, so future
Lance internals can't collide with a user column.
- **Vector columns carry dim + metric**, so an index can be built and
the metric round-trips on open without being re-specified.
- **Nested lists and lists-of-vectors rejected** rather than silently
mis-encoded.

### `generic_codec` — schema-driven encode/decode

JSON rows ↔ Arrow, matched **by name in both directions**. Two
deliberate choices:

- An undeclared key is an **error, not a silent drop** — a field-name
typo fails at the write instead of becoming missing data discovered much
later.
- Projected batches decode without their missing columns, which is what
lets a blob-excluding scan round-trip.

This also fixes by construction a bug class the built-in stores have:
they decode nested structs (`relationships`, `state_metadata`)
**positionally**, so reordering fields silently mis-assigns data.

### `generic_store` — the store

Delegates *all* storage behavior to `StorageBase`: resident writer,
fence retry, surgical WAL drain, compaction with `defer_index_remap`, id
index, LSM reads. `seal_on_add` picks rollout's deferred-seal profile
(default) or read-your-write.

## How users write it

```python
store = GenericStore.open(uri, schema={
    "id":        {"type": "string", "nullable": False},
    "user_id":   {"type": "string"},
    "score":     {"type": "float32"},
    "embedding": {"type": "vector", "dim": 768, "metric": "cosine"},
    "video":     {"type": "binary", "blob": True},
})

store.add([{"id": "r1", "user_id": "u1", "score": 0.9}])   # nullable cols may be omitted
store.list()                                  # no video — never materialized
store.get("r1", columns=["video"])            # fetch the blob for one row
```

## Blobs are inline, and that's deliberate

Per your point that this project exists to store big blobs: **inline
`LargeBinary`, never blob-v2 offload**. The LSM read path has no
blob-materialization step, so an offloaded column reads back as `None` —
which is exactly why `RolloutStore` stores artifacts inline today.

`blob: true` doesn't change *where* bytes live, it changes *who pays for
them*: blob columns are dropped from default scan projections, so `list`
never materializes them, and `get(id, columns)` fetches them per row.

I measured this path directly before designing around it:

| size | write | read | survives WAL merge |
|---|---|---|---|
| 5 MB | 80 ms | 42 ms | ✓ |
| 40 MB | 483 ms | 262 ms | ✓ |
| 120 MB | 1.39 s | 760 ms | ✓ |

The test suite pins a 4 MB round-trip through a merge (kept small so CI
stays fast).

⚠️ **Worth knowing for the 100 MB+ case:** `max_memtable_size` defaults
to 256 MB, so two such rows trigger a seal. That's not breakage — the
count/time merge triggers bound the resulting generations — but it does
raise seal frequency. `ShardWriterConfig` exposes the knob; no store
surfaces it yet. Follow-up if it bites.

## Schema is immutable in v1

Reopening with a different schema is **rejected**, not silently
reinterpreted over existing data. Evolution needs a versioning story and
is out of scope here.

## Verification

28 new tests: 8 spec validation, 9 codec round-trip, 11 store behavior —
including that duplicate ids dedup to the newest write (proving `id`
really is the merge key), deferred seal matches rollout's visibility
semantics, and generations merge into the base table.

209 core lib tests pass (up from 181), `cargo clippy --workspace
--all-targets` clean.

## Not in this PR

The server/client/Python surface — `CreateContextRequest` carrying a
schema, generic record DTOs, deduping the six duplicated conversion
functions, explicit column names in search/retrieve. This PR is the core
engine; the API layer is the next one, and it's large enough to want its
own review.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude <noreply@anthropic.com>

The three follow-ups from #220, plus a regression guard for a bug I
introduced while writing them.

## 1. `seal_on_add` is now persisted in the dataset

It lived only in `GenericStoreOptions`, so a store created with
`seal_on_add: true` **lost read-your-write the first time it was evicted
from the LRU and reopened** — the server passed a hardcoded `false` at
`state.rs:642`.

The failure mode is the nasty kind: no error, no data loss, just
"sometimes I can't read what I just wrote" — and triggered by *cache
pressure*, not by anything in the caller's code. Untestable locally,
only shows up under load. `/merge-wal` would accidentally paper over it
(it flushes internally), making it look intermittent.

The mode now sits in the schema metadata beside the spec, so a reopen
restores what the store was *created* with. Stores written before this
fall back to the caller's option, so nothing breaks on upgrade.

## 2. The sweepers now visit every store kind

Both `spawn_flush_sweeper` and `spawn_global_sweeper` hardcoded
`state.rollout_stores`. Consequences:

- **datagen**: seals on every append, so flush didn't matter — but
*nothing ever merged its generations*. They accumulated forever and
every read unioned all of them. This predates #217.
- **generic**: defaults to deferred seal, so writes stayed invisible
until someone called `/flush` by hand.

Rather than copy the loop a third time — the exact duplication #214
removed one layer down — the traversal is generic over a new `Sweepable`
trait in `sweeper.rs`.

**Why the trait is on `Arc<RwLock<Store>>` and not on the store:**
rollout's merge deliberately splits into a shared-lock prepare (the
expensive object-storage reads, during which appends keep flowing) and a
brief exclusive-lock commit. A trait over `&mut Store` would have forced
the exclusive lock across the whole merge and quietly stalled the write
path — a performance regression that no test would have caught. Letting
each kind own its locking preserves that.

Metric names keep their `rollout_` prefix and gain a `kind` label, so
existing dashboards and alerts keep working.

## 3. Config naming: documented, not renamed

`ROLLOUT_FLUSH_INTERVAL_SECS` and friends now govern all three kinds, so
the prefix is misleading. But renaming breaks deployment manifests for
zero functional gain, so I documented the actual scope in the flag docs
and the startup warning instead. Happy to add neutral aliases if you'd
rather.

## 4. ⚠️ A regression I caused, and the guard for it

While rewriting those doc comments my edit deleted two `#[arg(..)]`
attributes, turning `--rollout-flush-interval-secs` and
`--rollout-cleanup-interval-secs` into **required positional
arguments**:

```
error: the following required arguments were not provided:
  <ROLLOUT_CLEANUP_INTERVAL_SECS>
  <ROLLOUT_FLUSH_INTERVAL_SECS>
```

Everything still compiled. All 203 Rust tests still passed. The server
just refused to start. It surfaced only because the Python remote tests
spawn the real binary — and those tests swallow stderr, so it showed up
as "server did not become healthy".

Two tests in `config.rs` now close that hole: every field must be an
optional flag with a default, and the minimal documented invocation must
parse.

## Verification

I checked each fix actually fails without its fix, rather than trusting
a green run:

- reverting the persistence → both `seal_mode_survives_*` tests fail
- dropping the `#[arg]` → both config tests fail

9 new tests, including e2e ones that go through the real server path:
seal mode surviving an actual LRU eviction
(`generic_stores.lock().pop()`), and a sweeper pass draining a real
datagen generation.

Full suite: **203 core + 62 server + 18 api** Rust, **199 Python**.
clippy, ruff, pyright all clean.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude <noreply@anthropic.com>

… layers (#219)

## What

Completes the `DatagenStore` surface the datagen POC needs, wired
through all six layers (core → api → client/server/unified → PyO3 →
python wrapper) and exposed on the high-level
`lance_context.DatagenStore` wrapper so callers using the public Python
API reach every new function.

## Pieces

- **A1 — `DatagenStreamWriter`**: a client-side state machine that
stamps the bookkeeping columns (`item_seq`, `attempt`, `checkpoint_id`,
`event_id`) and returns event dicts to hand to
`append`/`append_checkpoint`. No new HTTP endpoint — works embedded and
remote. Fresh streams via `open_stream` (attempt=0, next_seq=1); resume
via `resume_stream` (attempt=`last_attempt+1`, continuing `item_seq`).
- **A3 — resume / attempt>0 fold semantics**: `resuming_writer` rebuilds
a writer from the folded item; the fold folds correctly across attempts.
- **A4 — `load_blob` by field name**: core `load_blob(folded,
field_name)` resolves the folded item's `blob_event_ids` map to bytes.
Propagated to Python via the folded dict's `blob_event_ids` +
`get_blob`, composed in the `api.py` wrapper.
- **A5 — inspection tree**: `item_tree` folds every projected descendant
to its latest state and links parent→child. Server does a thin raw
`events_for_root` dump; the client folds the tree via a single-source
`DatagenItemTree::build`.

## Python API

The high-level `lance_context.DatagenStore` wrapper gains `open_stream`
/ `resume_stream` / `item_tree` / `load_blob`, plus a
`DatagenStreamWriter` wrapper class (`step_started`, `step_completed`,
`item_terminal`, `item_failed`, `item_id`/`attempt` properties). Both
are re-exported from `lance_context`.

## Tests

- `python/tests/test_datagen.py` — 6 tests covering
open/resume/item_tree/load_blob/item_failed through the public wrapper.
- A core store integration test driving a real store: open a root
stream, complete a step with a Set field, checkpoint, terminal, spawn a
child stream, then assert `item_tree` links
roots/children/status/query_tags/fields.

All green: Rust 25 datagen core tests, Python 6 tests, `ruff`,
`pyright`, `cargo fmt --check`.

## Note

Branches from before origin/main merged the StorageBase migration
(#216/#217/#218). CI may surface conflicts against those; will
rebase/fix as needed.

---------

Co-authored-by: YangjunZ <yangjunzhang@microsoft.com>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>